//
// Created by zlxx on 2022/10/20.
//

#include <vector>
#include <algorithm>
#include "scan_selector.h"
#include "rc.h"
#include "sql/stmt/filter_stmt.h"

//! 这里只进行表内筛选
RC ScanSelector::scan(std::vector<std::vector<TupleValue>> &tuple_sets)
{
  RC rc = RC::SUCCESS;

  int count = 0;
  for (auto op : opers_) {
    tuple_sets.emplace_back();
    op->open();
    while (RC::SUCCESS == (rc = op->next())) {
      Tuple *tuple = op->current_tuple();
      if (nullptr == tuple) {
        rc = RC::INTERNAL;
        LOG_WARN("failed to get tuple from operator");
        break;
      }
      auto row_tp = dynamic_cast<RowTuple *>(tuple);
      TupleValue added_tuple;
      added_tuple.set_schema(row_tp->table(), row_tp->table()->table_meta().field_metas());
      added_tuple.set_record(row_tp->record());
      if (in_table_predicate(added_tuple)) {
        tuple_sets[count].push_back(added_tuple);
      }
    }
    count++;
    op->close();
  }
  return rc;
}

//! 表内筛选
bool ScanSelector::in_table_predicate(const TupleValue &tuple)
{
  if (filter_stmt_ == nullptr || filter_stmt_->filter_units().empty()) {
    return true;
  }
  for (const FilterUnit *filter_unit : filter_stmt_->filter_units()) {
    Expression *left_expr = filter_unit->left();
    Expression *right_expr = filter_unit->right();
    //! 存在别的表的值类型,肯定不是表内条件直接跳过
    if (left_expr->type() == ExprType::FIELD) {
      auto expr = dynamic_cast<FieldExpr &>(*left_expr);
      if (0 != strcmp(expr.table_name(), tuple.table()->name())) {
        continue;
      }
    }
    if (right_expr->type() == ExprType::FIELD) {
      auto expr = dynamic_cast<FieldExpr &>(*right_expr);
      if (0 != strcmp(expr.table_name(), tuple.table()->name())) {
        continue;
      }
    }
    CompOp comp = filter_unit->comp();
    TupleCell left_cell;
    TupleCell right_cell;
    left_expr->get_value(tuple, left_cell);
    right_expr->get_value(tuple, right_cell);

    bool filter_result = compare_cell(comp, left_cell, right_cell);
    if (!filter_result) {
      return false;
    }
  }
  return true;
}
bool ScanSelector::compare_cell(const CompOp &comp, const TupleCell &left_cell, const TupleCell &right_cell) const
{
  const int compare = left_cell.compare(right_cell);
  bool filter_result = false;
  switch (comp) {
    case EQUAL_TO: {
      filter_result = (0 == compare);
    } break;
    case LESS_EQUAL: {
      filter_result = (compare <= 0);
    } break;
    case NOT_EQUAL: {
      filter_result = (compare != 0);
    } break;
    case LESS_THAN: {
      filter_result = (compare < 0);
    } break;
    case GREAT_EQUAL: {
      filter_result = (compare >= 0);
    } break;
    case GREAT_THAN: {
      filter_result = (compare > 0);
    } break;
    case LIKE_MATCH: {
      filter_result = (left_cell.match_like(right_cell));
    } break;
    case NOT_LIKE_MATCH: {
      filter_result = !(left_cell.match_like(right_cell));
    } break;
    default: {
      LOG_WARN("invalid compare type: %d", comp);
    } break;
  }
  return filter_result;
}
RC ScanSelector::descartes(std::vector<std::vector<TupleValue>> &tuples, int position, std::vector<CompositeTuple> &res,
    CompositeTuple &a_line)
{
  if (position >= tuples.size()) {
    return RC::SUCCESS;
  }
  auto table_i_res = tuples[position];
  for (auto &table_i_re : table_i_res) {
    a_line.add_row_tuple(table_i_re);
    bool to_skip = false;
    between_table_predict_in_recursive(a_line, to_skip);
    if (to_skip) {
      a_line.remove_tail_tuple();
      continue ;
    }
    if (position == tuples.size() - 1) {
      bool to_add = false;
      between_table_predict(a_line, to_add);
      if (to_add) {
        const CompositeTuple &added(a_line);
        res.push_back(added);
      }
      a_line.remove_tail_tuple();
      continue;
    }
    RC rc = descartes(tuples, position + 1, res, a_line);
    if (rc != RC::SUCCESS) {
      return rc;
    }
    a_line.remove_tail_tuple();
  }
  return RC::SUCCESS;
}

RC ScanSelector::get_final_result(const std::vector<CompositeTuple> &tuples, std::vector<CompositeTuple> &res)
{
  for (const auto &tuple : tuples) {
    bool predict = false;
    RC rc = between_table_predict(tuple, predict);
    if (rc != RC::SUCCESS) {
      return rc;
    }
    if (predict) {
      res.push_back(tuple);
    }
  }
  return RC::SUCCESS;
}

//! 在递归过程中 如果发现有可以筛选的值,直接筛选,
//! 如果发现没出现的字段也在比较，那也不能说明发生了错误
//! 错误要在最后一边大筛中决定
void ScanSelector::between_table_predict_in_recursive(const CompositeTuple &tuple, bool &to_stop) const
{
  std::vector<const Table *> tables;
  for (const auto &t : tuple.tuples()) {
    tables.push_back(t.table());
  }
  for (auto filter_unit : filter_stmt_->filter_units()) {
    if (filter_unit->right()->type() == ExprType::VALUE || filter_unit->left()->type() == ExprType::VALUE) {
      continue;
    }
    Expression *left_expr = filter_unit->left();
    Expression *right_expr = filter_unit->right();
    const auto left_field = dynamic_cast<FieldExpr *>(left_expr);
    const auto right_field = dynamic_cast<FieldExpr *>(right_expr);
    if (std::find(tables.begin(), tables.end(), left_field->field().table()) == tables.end() ||
        std::find(tables.begin(), tables.end(), right_field->field().table()) == tables.end()) {
      continue;
    }
    TupleCell left_cell;
    TupleCell right_cell;
    tuple.find_cell(left_field->field(), left_cell);
    tuple.find_cell(right_field->field(), right_cell);
    if (!compare_cell(filter_unit->comp(), left_cell, right_cell)) {
      to_stop = true;
      break ;
    }
  }
}

RC ScanSelector::between_table_predict(const CompositeTuple &tuple, bool &ret) const
{
  for (const FilterUnit *filter_unit : filter_stmt_->filter_units()) {
    Expression *left_expr = filter_unit->left();
    Expression *right_expr = filter_unit->right();
    if (left_expr->type() == ExprType::VALUE || right_expr->type() == ExprType::VALUE) {
      continue;
    }
    const auto left_field = dynamic_cast<FieldExpr *>(left_expr);
    const auto right_field = dynamic_cast<FieldExpr *>(right_expr);
    TupleCell left_cell;
    TupleCell right_cell;
    RC rc1 = tuple.find_cell(left_field->field(), left_cell);
    RC rc2 = tuple.find_cell(right_field->field(), right_cell);
    if (rc1 != RC::SUCCESS || rc2 != RC::SUCCESS) {
      return RC::SCHEMA_FIELD_NOT_EXIST;
    }
    if (!compare_cell(filter_unit->comp(), left_cell, right_cell)) {
      ret = false;
      return RC::SUCCESS;
    }
  }
  ret = true;
  return RC::SUCCESS;
}

RC ScanSelector::get_results(std::vector<std::vector<TupleValue>> &tuples, std::vector<CompositeTuple> &res)
{
  RC rc = RC::SUCCESS;
  auto line = CompositeTuple();
  rc = descartes(tuples, 0, res, line);
  std::vector<CompositeTuple> temp;
  rc = get_final_result(res, temp);
  if (rc != RC::SUCCESS) {
    return rc;
  }
  res.swap(temp);
  return RC::SUCCESS;
}
